package cn.doitedu.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job02_每5分钟各类页面访问人数最多的前10个页面 {

    public static void main(String[] args) {

        // 编程环境
        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射kafka中dwd层的行为日志topic: dwd-events
        tenv.executeSql(
                "create table dwd_events_kafka(\n" +
                "     session_id         string\n" +
                "    ,event_id           string\n" +
                "    ,action_time        bigint\n" +
                "    ,user_id            bigint\n" +
                "    ,page_type          string\n" +
                "    ,properties         map<string,string>\n" +
                "    ,url as properties['url']     "+
                "    ,rt as to_timestamp_ltz(action_time,3)         \n" +
                "    ,watermark for rt as rt - interval '0' second  \n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'dwd-events',\n" +
                "    'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "    'properties.group.id' = 'doit44_g1',\n" +
                "    'scan.startup.mode' = 'latest-offset',\n" +
                "    'value.format' = 'json',\n" +
                "    'value.fields-include' = 'EXCEPT_KEY'\n" +
                ")                                                    ");


        // 统计sql
        // 每5分钟各类页面访问人数最多的前10个页面
        tenv.executeSql("with tmp as (\n" +
                "    SELECT\n" +
                "        *\n" +
                "    from dwd_events_kafka\n" +
                "    where event_id = 'page_load'\n" +
                ")\n" +
                "\n" +
                "\n" +
                "SELECT \n" +
                "   *\n" +
                "FROM (\n" +
                "    SELECT\n" +
                "        window_start,\n" +
                "        window_end,\n" +
                "        page_type,\n" +
                "        url,\n" +
                "    \tuv,\n" +
                "    \trow_number() over(partition by window_start,window_end,page_type order by uv desc) as rn\n" +
                "    FROM (\n" +
                "        SELECT\n" +
                "            window_start,\n" +
                "        \twindow_end,\n" +
                "        \tpage_type,\n" +
                "        \turl,\n" +
                "        \tcount(distinct user_id) as uv\n" +
                "        from table(\n" +
                "            tumble(table tmp,descriptor(rt),interval '5' minute)\n" +
                "        )\n" +
                "        group by \n" +
                "            window_start,\n" +
                "        \twindow_end,\n" +
                "        \tpage_type,\n" +
                "        \turl\n" +
                "    ) o1\n" +
                ") o2\n" +
                "where rn<=10");








    }


}
